/*
 * Copyright 2023-2024 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.icodening.collapse.core;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;

/**
 * @author icodening
 * @date 2023.05.23
 */
public abstract class CollapseExecutorBlockingSupport<INPUT, OUTPUT, BATCH_OUTPUT> extends AbstractCollapseExecutor<INPUT, OUTPUT, BATCH_OUTPUT> {

    public CollapseExecutorBlockingSupport() {
        super();
    }

    public CollapseExecutorBlockingSupport(ListeningCollector collector) {
        super(collector);
    }

    /**
     * 这个方法是受保护的，返回类型是 OUTPUT，并且接受一个 Bundle<INPUT, OUTPUT> 类型的参数。这个方法可能会抛出 Throwable。
     *
     * 这种方法适用于需要同步等待异步结果的场景，同时确保资源被正确清理。
     * 然而，它也可能导致线程阻塞，特别是在高并发或长时间等待的情况下。
     * @param bundle
     * @return
     * @throws Throwable
     */
    @Override
    protected final OUTPUT returning(Bundle<INPUT, OUTPUT> bundle) throws Throwable {
        ThreadlessExecutor threadlessExecutor = (ThreadlessExecutor) bundle.getCallbackExecutor();
        CompletableFuture<OUTPUT> completableFuture = bundle.getListeningResult();
        try {
            //方法中使用了一个 while 循环，通过调用 threadlessExecutor.waitAndDrain() 来等待 CompletableFuture 完成。
            // 如果 CompletableFuture 还没有完成，waitAndDrain 方法会阻塞当前线程，直到 CompletableFuture 完成或有新的任务可以执行。
            while (!completableFuture.isDone()) {
                threadlessExecutor.waitAndDrain();
            }
            //一旦 CompletableFuture 完成，使用 completableFuture.get() 获取结果。
            // 这个方法会阻塞直到 CompletableFuture 完成，并返回其结果。
            return completableFuture.get();
        } catch (ExecutionException executionException) {
            //throw actual exception
            throw executionException.getCause();
        } finally {
            threadlessExecutor.shutdown();
        }
    }

    /**
     * 使用默认的回调线程池，有所风险
     * @return
     */
    @Override
    protected final Executor getCallbackExecutor() {
        return new ThreadlessExecutor();
    }
}
